package com.shujia.flink.core

import org.apache.flink.streaming.api.scala._

object Demo3Parallelism {
  def main(args: Array[String]): Unit = {


    //1、创建flink的运行环境,相当于sparkContext
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    /**
     * flink并行度设置（默认是1），并行度越高，每秒能处理的数据量越多
     * 1、env.setParallelism(2):在代码的前面统一设置
     * 2、-p 3 ：提交任务时设置
     * 3、每一个算子可以单独设置并行度
     *
     * 优先级 3 > 1 > 2
     * 在企业中一般使用第二种，比较灵活，不需要重新打包
     *
     *
     * 并行度设置原则  -- 由数据量决定（每秒的数据量--- 吞吐量）
     * 1、非聚合聚合计算时（计算效率高） -- 每个并行度可以处理的数据量是（每秒 （100000））
     * 1、聚合聚合计算时（计算效率低） -- 每个并行度可以处理的数据量是（每秒 （10000））
     *
     */

    //env.setParallelism(4)


    //2、读取数据
    //从socket中读取数据,nc -lk 8888
    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    //3、统计单词的数量
    //一行转多行
    val wordsDS: DataStream[String] = linesDS
      .flatMap(line => line.split(","))
      .setParallelism(5)


    //转换成kv格式
    val kvDS: DataStream[(String, Int)] = wordsDS.map(word => (word, 1))

    //按照单词分组
    val keyByDS: KeyedStream[(String, Int), String] = kvDS.keyBy(kv => kv._1)

    //对下标为1的字段求和
    val countDS: DataStream[(String, Int)] = keyByDS.sum(1)

    //打印结果
    countDS.print()

    //启动flink任务
    env.execute()
  }

}
